package com.leilei.join.interval;

import com.leilei.join.common.Vehicle;
import com.leilei.join.common.VehicleDetail;
import com.leilei.join.common.VehicleInfo;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.util.Collector;

/**
 * @author lei
 * @version 1.0
 * @date 2021/3/27 22:00
 * @desc 车辆与车辆明细Join 关联处理
 */
public class VehicleJoinDetailProcessFunction extends ProcessJoinFunction<Vehicle, VehicleDetail, VehicleInfo> {
    @Override
    public void processElement(Vehicle left, VehicleDetail right, Context ctx, Collector<VehicleInfo> out) throws Exception {
        out.collect(VehicleInfo.builder()
                .vehicleId(left.getId())
                .color(right.getColor())
                .createTime(right.getCreateTime())
                .type(left.getType())
                .build());
    }
}
